package bolts;

import java.io.File;
import java.io.FileNotFoundException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Scanner;
import java.util.Set;

import output.TopWord;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import common.WordFrequency;
import common.WordSpaceSavingAlgorithm;

public class DetectTopWordsBolt extends AbstractTimeWindowBolt {

    private int topWordsOutputSize;

    private WordSpaceSavingAlgorithm topWordsDetector;

    private final Set<String> filteredWords = new HashSet<String>();

    public DetectTopWordsBolt(String boltName, Map stormConf) {
        super(boltName, stormConf);
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        super.prepare(stormConf, context);

        this.topWordsOutputSize = Integer.valueOf((String) stormConf.get("topWordsOutputSize"));
        this.topWordsDetector = new WordSpaceSavingAlgorithm(Integer.valueOf((String) stormConf.get("topWordsSpaceSavingK")));

        try {
            final Scanner scanner = new Scanner(new File((String) stormConf.get("filteredWords")));
            while (scanner.hasNextLine()) {
                filteredWords.add(scanner.nextLine().trim().toLowerCase());
            }
            scanner.close();
        } catch (FileNotFoundException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("topWords"));
    }

    @Override
    protected void handleNormalTuple(Tuple input, BasicOutputCollector collector) {
        final String[] tokens = input.getStringByField("text").toLowerCase().split(" ");
        for (final String token : tokens) {
            if (!filteredWords.contains(token)) {
                topWordsDetector.processNewWord(token);
            }
        }

        final Set<String> filteredTokens = new HashSet<String>(Arrays.asList(tokens));
        filteredTokens.removeAll(filteredWords);
        for (final String token : filteredTokens) {
            topWordsDetector.incrementTweetCount(token);
        }
    }

    @Override
    protected void handleTickTuple(Tuple input, BasicOutputCollector collector) {
        final List<TopWord> output = new LinkedList<TopWord>();
        for (final WordFrequency wf : topWordsDetector.getTopItems(topWordsOutputSize)) {
            output.add(new TopWord(wf.getWord(), wf.getTweetCount(), (double) wf.getCount() / timeWindowInSeconds));
        }
        collector.emit(new Values(output));
        topWordsDetector.reset();
    }

}
